-- created by kumiler
-- on 2022/7/26 9:12
-- DESC 车件分离明细表DDL
-- modify v1.1     : jimmyk.ji
-- modify time : 2023-04-28 车件分离统计增加8种剔除类型数据，不需要参与车件分离判断；优化之前的逻辑
-- modify v1.2     : jimmyk.ji
-- modify time : 2023-05-12 优化剔除规则6的逻辑

with net as (
    select name,
           code,
           parent_network_code,                    -- 加盟商/中心：代理区；网点：加盟商
           parent_network_name,
           nvl(agent_code,financial_center_code) as agent_code,
           nvl(agent_name,financial_center_desc) as agent_name,
           manage_code  as regional_id,   -- bigint comment '大区ID',
           manage_name  as regional_desc, -- string comment '大区名称',
           network_type,
           if(network_type = 4,1,0) as is_distribution_center
    from jms_dim.dim_network_whole_massage
    where instr(name, '测试') = 0
      and instr(code, 'test') = 0
),
     shipment_msg as (   -- 2
         select shipment_no,
                max(shipment_name) as shipment_name,
                min(actual_arrival_time) as actual_arrival_time,
                max(actual_departure_time) as actual_departure_time,
                max(vehicleline_code) as line_code, -- 车线code
                max(vehicleline_name) as line_name,
                max(start_code) as start_code,
                max(end_code) as end_code,
                1 as type1
               from jms_dwd.dwd_tmsnew_shipment_union_base_dt
               where dt >= date_add('{{ execution_date | cst_ds }}',-15)
               and dt <= '{{ execution_date | cst_ds }}'
               group by shipment_no
     ),

     shipment_stop_msg as (   -- 6         干线经停信息
        select  shipment_no
               ,network_id
               ,network_code
               ,max(network_name) as network_name
          from jms_dwd.dwd_tmsnew_shipment_stop_dt
         where dt >= date_add('{{ execution_date | cst_ds }}',-15)
           and dt <= '{{ execution_date | cst_ds }}'
           and sort_num>1
         group by shipment_no
               ,network_id
               ,network_code
     ),


     waybill_source as (
         select waybill_no as billcode,
                ordersource_code,
                ordersource_name,
                goods_type_code,
                goods_type_name
         from jms_dwd.dwd_yl_oms_oms_waybill_incre_dt
         where dt >= date_add('{{ execution_date | cst_ds }}', -15)
         and dt <= '{{ execution_date | cst_ds }}'
           and is_delete <> 2
           and (waybill_no like 'JT%' or waybill_no like '55%')
     ),
-- 获取中心卸车扫描 T-15 12:00:00 -- T-1 12:00:00 的数据
     unloading as (
         select a.*
         from (
                  select t0.waybill_no as billcode,       -- string comment '运单编号',
                         t0.shipment_no as transfercode,   -- string comment '转单号',
                         s0.shipment_name,
                         s0.actual_arrival_time,
                         s0.line_code,
                         s0.line_name,
                         s0.start_code,
                         s0.end_code,
                         scan_type as scantype,       -- string comment '扫描类型',
                         transfer_center as transfercenter, -- string comment '中转中心',
                         pre_station as nextstation,    -- string comment '上一站或下一站',
                         scan_site_code as scansitecode,   -- string comment '扫描网点编号',
                         scan_time as scantime,       -- timestamp,扫描时间
                         pre_station_code as nextstationcode,
                         goods_type as goodstype,
                         row_number()
                                 over (partition by scan_site_code, waybill_no order by scan_time) as scantime_rank
                  from jms_dwd.dwd_tab_barscan_unloading_base_dt t0
                  left join shipment_msg s0 on t0.shipment_no = s0.shipment_no
                  -- 别问为啥是15天，产品提的
                  where t0.dt >= date_add('{{ execution_date | cst_ds }}', -15)
                  and t0.dt <= '{{ execution_date | cst_ds }}'
                    and scan_type = '卸车扫描'
                    and (waybill_no like 'JT%' or waybill_no like '55%')
                    and t0.shipment_no is not null
              ) a
                  inner join
                  (select * from net where is_distribution_center = 1) n
                  on a.scansitecode = n.code
         where a.scantime_rank = 1
     ),
-- 获取装车扫描取最晚 T-8 12:00:00天的所有数据
     loading as (
         select *
         from (
                  select t1.waybill_no as billcode,        -- string,运单编号
                         t1.shipment_no as transfercode,    -- string,转单号
                         s1.shipment_name,
                         s1.actual_arrival_time,
                         s1.actual_departure_time,
                         s1.line_code,
                         s1.line_name,
                         scan_type as scantype,        -- string,扫描类型
                         transf_ercenter as transfercenter,  -- string comment '中转中心',
                         next_station as nextstation,     -- string,上一站或下一站
                         scan_site_code as scansitecode,    -- string comment '扫描网点编号',
                         scan_time as scantime,        -- timestamp,扫描时间
                         next_station_code as nextstationcode, -- string,上下一站编号
                         goods_type as goodstype,
                         row_number()
                                 over (partition by scan_site_code, waybill_no order by scan_time desc) as scantime_rank
                  from jms_dwd.dwd_tab_barscan_loading_base_dt t1
                  left join shipment_msg s1 on t1.shipment_no = s1.shipment_no
                  -- 取较卸车表早三天的数据
                  where t1.dt >= date_add('{{ execution_date | cst_ds }}', -20) -- 向前取8天
                  and t1.dt <= '{{ execution_date | cst_ds }}'
                    and t1.scan_type = '装车扫描'
                    and (t1.waybill_no like 'JT%' or t1.waybill_no like '55%')
                    and t1.shipment_no is not null
              ) a
         where scantime_rank = 1
     ),
         merge1 as ( -- jms_tmp.ttmp08251
         select unloading.billcode,
                unloading.scantime,
                unloading.scantype,
                unloading.scansitecode,
                unloading.transfercode                                      as unload_shipment_no,
                unloading.shipment_name                                     as unload_shipment_name,
                unloading.actual_arrival_time                               as unload_actual_arrival_time,
                unloading.line_code,
                unloading.line_name,
                unloading.start_code as unloading_start_code,
                unloading.nextstationcode,
                loading.nextstationcode                                     as load_station_lastcode,
                loading.transfercode                                        as load_shipment_no,
                loading.shipment_name                                       as load_shipment_name,
                loading.scantime                                            as load_scantime,
                loading.scansitecode                                        as load_station_code,
                lead(unloading.scantime, 1) over (partition by unloading.scansitecode, loading.transfercode order by unloading.scantime) as next_scantime
         from unloading
                  left join loading on unloading.billcode = loading.billcode and
                                       unloading.scansitecode = loading.nextstationcode
             and unloading.scantime > loading.scantime
         where loading.transfercode is not null
     ),

     -- 问题件
     problem as ( -- 1 添加剔除的问题件   4
         select waybill_no,transfer_code as shipement_no,4 as type1
         from jms_dwd.dwd_tab_barscan_difficult_base_dt
         where dt >= date_add('{{ execution_date | cst_ds }}',-9)
         and dt <= '{{ execution_date | cst_ds }}'
         group by waybill_no,transfer_code
     ),
     -- 获取卸车扫描时间 > 本站装车扫描的运单  3
     wrong_time as (
         select unloading.billcode,unloading.transfercode,3 as type1
         from unloading
         left join loading
         on unloading.billcode = loading.billcode
         and unloading.scansitecode = loading.scansitecode
         where unloading.scantime > loading.scantime
         group by unloading.billcode,unloading.transfercode
     ),
     -- 剔除 卸车到件扫描任务单的起始站点 <> 装车扫描的扫描站点   1
     wrong_station as (
         select billcode,unload_shipment_no,1 as type1
         from merge1
         where unloading_start_code <> load_station_code
         group by billcode,unload_shipment_no
     ),
     -- 剔除 本站卸车扫描时间 小于上一站装车扫描时间的运单  5
     wrong_time1 as (
         select unloading.billcode,unloading.transfercode,5 as type1
         from unloading
         left join loading
         on unloading.billcode = loading.billcode
         and unloading.scansitecode = loading.nextstationcode
         where unloading.scantime <= loading.actual_departure_time
         group by unloading.billcode,unloading.transfercode
     ),
      -- 剔除 同一站点先发后到的运单  7
     wrong_time2 as (
         select unloading.billcode,unloading.transfercode,7 as type1
         from unloading
         left join loading
         on unloading.billcode = loading.billcode
         and unloading.scansitecode = loading.scansitecode
         where unloading.scantime > loading.scantime
         group by unloading.billcode,unloading.transfercode
     ),
     -- 剔除 上一站多次建包  8
     wrong_bagging as(
     select waybill_no,
      8 as type1
      from (
      select waybill_no,scan_site_code,count(1) as cnt
      from jms_dwd.dwd_tab_barscan_bagging_base_dt
      where dt>=date_add('{{ execution_date | cst_ds }}',-20)
      and dt<='{{ execution_date | cst_ds }}'
      and scan_type='建包扫描'
      group by waybill_no,scan_site_code
    )tmp
where tmp.cnt>1
    group by waybill_no
         ),
         --剔除多次揽收
          barscan_taking_base as(
          select waybill_no,8 as type1
               from (
               select waybill_no,scan_site_code,count(1) as cnt
               from jms_dwd.dwd_tab_barscan_taking_base_dt
               where dt>=date_add('{{ execution_date | cst_ds }}',-20)
                 and dt<='{{ execution_date | cst_ds }}'
               group by waybill_no,scan_site_code
             )tmp
         where tmp.cnt>1
             group by waybill_no
           ),

       -- 剔除 卸车任务站点和装车任务下一站不一样的   6
     wrong_station_2 as (
         select unloading.billcode,unloading.transfercode,6 as type1
         from unloading
         left join loading
           on unloading.billcode = loading.billcode
          and unloading.scansitecode=loading.nextstationcode
         left join shipment_stop_msg   ---经停站点
           on unloading.transfercode = shipment_stop_msg.shipment_no    ---任务号
          and unloading.scansitecode = shipment_stop_msg.network_code   ---卸车扫描站点 = 经停站点
        where unloading.start_code = loading.scansitecode
          and shipment_stop_msg.network_code is null
         --and unloading.scansitecode <> unloading.end_code
         -- where unloading.actual_arrival_time <= loading.actual_departure_time
         group by unloading.billcode,unloading.transfercode
     ),

-- 补充卸车网点信息 & 是否超3小时 & 中心数量 & 全网数量 & 装卸车网点数据

     unload_agg as(
         select scansitecode, billcode_center_cnt,
                sum(billcode_center_cnt) over() as billcode_cnt
         from
              (select scansitecode, count(1) as billcode_center_cnt
                 from unloading
                 where scantime >= concat(date_add('{{ execution_date | cst_ds }}',-1), ' 12:00:00')
                   and scantime < concat('{{ execution_date | cst_ds }}' ,' 12:00:00')
                 group by scansitecode
                ) agg1
     ),

    rank_tab as
          (select merge1.*
                  ,nvl(merge1.unload_shipment_name,ship1.shipment_name)  as unload_shipment_name1
                  ,nvl(merge1.load_shipment_name,ship2.shipment_name)    as load_shipment_name1
                  ,unload_agg.billcode_cnt                               as billcode_cnt
                  ,unload_agg.billcode_center_cnt                        as billcode_center_cnt
                  ,nvl(coalesce(wrong_station.type1,if(ship1.shipment_no is null,2,null),wrong_time.type1,problem.type1,wrong_time1.type1,wrong_station_2.type1,wrong_time2.type1,nvl(wrong_bagging.type1,barscan_taking_base.type1  )),0)
                   as delete_type
           from merge1
           left join problem    -- 4
           on merge1.billcode = problem.waybill_no
           left join wrong_station    -- 1
           on merge1.billcode = wrong_station.billcode and merge1.unload_shipment_no = wrong_station.unload_shipment_no
           left join wrong_time    -- 3
           on merge1.billcode = wrong_time.billcode and merge1.unload_shipment_no = wrong_time.transfercode
           left join shipment_msg ship1 on merge1.unload_shipment_no = ship1.shipment_no  -- 2
           left join shipment_msg ship2 on merge1.load_shipment_no = ship2.shipment_no
           left join unload_agg on merge1.scansitecode = unload_agg.scansitecode
           left join wrong_time1 on merge1.billcode = wrong_time1.billcode and merge1.unload_shipment_no = wrong_time1.transfercode    -- 5
           left join wrong_station_2 on merge1.billcode=wrong_station_2.billcode and merge1.unload_shipment_no=wrong_station_2.transfercode   -- 新加   6
           left join  wrong_time2 on merge1.billcode=wrong_time2.billcode and merge1.unload_shipment_no=wrong_time2.transfercode           --新加   7
           left join wrong_bagging on  merge1.billcode=wrong_bagging.waybill_no   --新加  8
           left join barscan_taking_base on merge1.billcode=barscan_taking_base.waybill_no
           ),

 rank_tab1 as
          (select *,
                abs(round((to_unix_timestamp(lead(rank_tab.scantime, 1)
                                                  over (partition by rank_tab.scansitecode, rank_tab.load_shipment_no order by rank_tab.scantime))
                    -
                           to_unix_timestamp(rank_tab.scantime)) / 3600, 2)) as diff_timestamp,    ---下一单与当前运单的扫描时间差
                abs(round((to_unix_timestamp(rank_tab.scantime)
                    -
                           to_unix_timestamp(lag(rank_tab.scantime,1,null) over(partition by rank_tab.scansitecode, rank_tab.load_shipment_no order by rank_tab.scantime) )) / 3600, 2))
                    as diff_scantime                                                              ---当前运单与上一单的扫描时间差

          from rank_tab
        where delete_type  = 0
        union all

        select *,
                null as diff_timestamp,
                null as diff_scantime

          from rank_tab
        where delete_type  <> 0

          ),

     rank_tab2 as (select *,
                  if(diff_timestamp >= 3,
                     row_number() over (partition by load_shipment_no,scansitecode order by scantime),
                     null) as scan_time_rank
           from rank_tab1),

    targe as ( -- 超过3h未扫描的第一个运单
        select scansitecode, load_shipment_no, min(scantime) as scantime
        from rank_tab2
        where scan_time_rank is not null
        group by scansitecode, load_shipment_no
     ) ,
     res as (
         select tab.billcode
                ,tab.scantime
                ,tab.scantype
                ,tab.scansitecode as unload_scansite_code
                ,tab.unload_shipment_no
                ,tab.unload_shipment_name
                ,unload_actual_arrival_time
                ,line_code,line_name
                ,tab.nextstationcode
                ,tab.load_shipment_no
                ,tab.load_shipment_name
                ,tab.load_station_code
                ,tab.scan_time_rank
                ,tab.next_scantime
                ,tab.diff_timestamp
                ,targe.scantime as timeout_scantime
                ,tab.diff_scantime
                ,tab.unloading_start_code
                ,tab.unload_shipment_name1     --add jixk 20230420
                ,tab.load_shipment_name1       --add jixk 20230420
                ,tab.billcode_cnt              --add jixk 20230420
                ,tab.billcode_center_cnt       --add jixk 20230420
                ,tab.delete_type               --add jixk 20230420
         from rank_tab2 tab
         left join targe on tab.scansitecode = targe.scansitecode and tab.load_shipment_no = targe.load_shipment_no
         where tab.unload_shipment_no <> tab.load_shipment_no
            or tab.scantime >= targe.scantime
     ),
res1 as (
    select res.billcode                                              as bill_code               -- 运单号
         , goods_type_code                                                                      -- 物品类型code
         , goods_type_name                                                                      -- 物品类型名称
         , ordersource_code                                                                     -- 订单来源code
         , nvl(ordersource_name, '其他')                               as ordersource_name        -- 订单来源
         , res.unload_shipment_no                                                                     -- 任务号
         , res.unload_shipment_name
         , scantime                                                  as scan_time               -- 卸车扫描时间
         , next_scantime                                                                        -- 下一单扫描时间',
         , diff_timestamp      as nextbill_scantime_diff                                        -- 与下一单扫描时间 间隔-分钟
         , load_station_code                                         as unload_last_stationcode -- 卸车上一站code
         , net1.name                                                 as unload_last_station     -- 卸车上一站
         , unload_scansite_code                                      as transfer_center         -- 卸车转运中心
         , net2.name                                                 as transfer_center_name    -- 卸车转运name
         , unload_actual_arrival_time                                                  -- 实际到达时间
         , line_code                                                            -- 车线code
         , line_name                                                            -- 车线name
         , net2.regional_id                                          as unload_regional_id      -- 卸载中心大区id
         , net2.regional_desc                                        as unload_regional_desc    -- 卸载中心大区名
         , net2.agent_code                                           as unload_agent_code       -- 卸载中心代理区code
         , net2.agent_name                                           as unload_agent_name       -- 卸载中心代理区
         , ''                                                        as unload_franchisee_code  -- 卸车中心加盟商code
         , ''                                                        as unload_franchisee_name  -- 卸车中心加盟商
         , net1.code                                                 as load_station_code       -- 装车网点
         , net1.name                                                 as load_station_name       -- 装车网点
         , net1.regional_id                                          as load_regional_id        -- 装车中心大区id
         , net1.regional_desc                                        as load_regional_desc      -- 装车中心大区名
         , net1.agent_code                                           as load_agent_code         -- 装车中心代理区code
         , net1.agent_name                                           as load_agent_name         -- 装车中心代理区
         , if(net1.network_type = 6, net1.parent_network_code, null) as load_franchisee_code    -- 装车所属网点加盟商code
         , if(net1.network_type = 6, net1.parent_network_name, null) as load_franchisee_name    -- 装车所属网点加盟商
         , net1.is_distribution_center                               as center_load             -- 是否中心装车
         , res.load_shipment_no                                                                 -- 装车上一站任务号
         , res.load_shipment_name
         , scan_time_rank                  as  scan_time_rank                                   -- 同一任务号中间隔时间大于3h的排名
         , timeout_scantime                                                                     -- 首次超时扫描的的时间
         , if(unload_shipment_no <> load_shipment_no, 1, 0)         as diff_shipment            -- 装卸任务单是否相同
         , if(scantime > timeout_scantime, 1, 0)                    as timeout_3h               -- 扫描是否超三小时
         , case
               when unload_shipment_no <> load_shipment_no
                   and scantime > timeout_scantime
                   then 1
               else 0 end                                            as bill_separated          -- tinyint comment '是否车件分离',
         , date_format(`current_timestamp`(), 'yyyy-MM-dd HH:mm:ss') as etl_insert_time         -- string comment '数据清洗写入时间 current_time',
         , date_add('{{ execution_date | cst_ds }}',-1)                                 as date_time               -- string comment '数据时间【与dt同值，导入doris使用】'
         ,diff_scantime
        ,res.unloading_start_code
        ,res.unload_shipment_name1
        ,res.load_shipment_name1
        ,res.billcode_cnt
        ,res.billcode_center_cnt
        ,res.delete_type
    from res
        left join net net1 on res.load_station_code = net1.code
        left join net net2 on res.unload_scansite_code = net2.code
        left join waybill_source source on res.billcode = source.billcode
    where scantime >= concat(date_add('{{ execution_date | cst_ds }}',-1), ' 12:00:00')
              and scantime < concat('{{ execution_date | cst_ds }}' ,' 12:00:00')
)

insert overwrite table jms_dm.dm_waybill_transfer_separated_detail_dt partition ( dt)
select
 res1.bill_code
,res1.goods_type_code
,res1.goods_type_name
,res1.ordersource_code
,res1.ordersource_name
,res1.unload_shipment_no
,res1.unload_shipment_name1 as unload_shipment_name
,res1.scan_time
,res1.next_scantime
,res1.nextbill_scantime_diff
,res1.unload_last_stationcode
,res1.unload_last_station
,res1.transfer_center
,res1.transfer_center_name
,res1.unload_actual_arrival_time
,res1.line_code
,res1.line_name
,res1.unload_regional_id
,res1.unload_regional_desc
,res1.unload_agent_code
,res1.unload_agent_name
,res1.unload_franchisee_code
,res1.unload_franchisee_name
,res1.load_station_code
,res1.load_station_name
,net.network_type as load_station_type
,res1.load_regional_id
,res1.load_regional_desc
,res1.load_agent_code
,res1.load_agent_name
,res1.load_franchisee_code
,res1.load_franchisee_name
,res1.center_load
,res1.load_shipment_no
,res1.load_shipment_name1  as load_shipment_name
,res1.scan_time_rank
,res1.billcode_cnt
,res1.billcode_center_cnt
,res1.timeout_scantime
,res1.diff_shipment
,res1.timeout_3h
,if(res1.delete_type >0,0,bill_separated)  as bill_separated
,res1.etl_insert_time
,res1.date_time
,diff_scantime
,delete_type
,date_add('{{ execution_date | cst_ds }}',-1) as dt
from res1
left join net on res1.load_station_code = net.code
where res1.diff_shipment = 1 or res1.timeout_3h = 1 or nextbill_scantime_diff >= 3
distribute by dt ,pmod(hash(bill_code),20)
;
